#include "xworker_single_queue.h"
#include "xtimer.h"
#include "xkillable_task.h"
#include "xassert.h"
#include "xthread_manager.h"
#include "xdebug.h"

namespace xge{
namespace xge_internal{

  size_t WorkerSingleQueue::Dequeue( TaskPtrVector* container , size_t size ) {

    const size_t qsize = queue_.size();
    if(size ==0) {
      size=qsize;
    } else {
      size=size>qsize?qsize:size;
    }
    // do the preparation for the container
    container->clear();
    if( container->capacity () < size ){
      container->reserve(size);
    }
    // do the enqueue stuff here
    queue_sync_lock_.Lock();
    size_t rsize = 0;
    QueueType::iterator ib = queue_.begin() , ie = queue_.end();
    while( size >0 && ib != ie ) {
      (*ib).second->set_task_status( Task::TASK_STATUS_LOCKING );
      container->push_back((*ib).second);
      ib = queue_.erase(ib);
      --size;
      ++rsize;
    }
    queue_sync_lock_.Unlock();
    return rsize;
  }

  bool WorkerSingleQueue::PushTask( const TaskPtr& task_ptr ) {
    if(  status() == WorkerQueue::STATUS_LOCKING || status_atomic () ==WorkerQueue::STATUS_RELEASE ) {
        return false;
    }

    task_ptr->set_task_status( Task::TASK_STATUS_WAITING );
    // push the task ptr into the queue
    KillableTask* killable_task_ptr = task_ptr->DownCast<KillableTask>();
    queue_sync_lock_.Lock();
    std::pair< QueueType::iterator , bool > ret =queue_.insert(
      QueueType::value_type(killable_task_ptr->name(),task_ptr)
      );
    queue_sync_lock_.Unlock();
    enqueue_times_total_++;
    // do some wake up stuff here
    if( !ret.second ) {
      return false;
    } else {
      WakeUp();
      return true;
    }
  }

  size_t WorkerSingleQueue::PushTask( TaskPtrList* task_list ) {
    if(  status() == WorkerQueue::STATUS_LOCKING || status_atomic () ==WorkerQueue::STATUS_RELEASE ) {
        return false;
    }
    // batching the task into the list 
    size_t push_size=0;
    AutoLock<FastLock> l(queue_sync_lock_);
    TaskPtrList::iterator ib = task_list->begin() , ie = task_list->end();
    while( ib != ie ) {
      // down cast the pointer into the killable task ptr type
      KillableTask* killable_task_ptr = (*ib)->DownCast<KillableTask>();
      killable_task_ptr->set_task_status( Task::TASK_STATUS_WAITING );
      // insert the ptr into the map
      std::pair< QueueType::iterator , bool > ret =queue_.insert(
        QueueType::value_type(killable_task_ptr->name(),(*ib))
        );
      // testing insertion correct or not
    if( ret.second )
      ++push_size;
    // update the iterator
    ++ib;
    }
    enqueue_times_total_+=push_size;
    if( push_size !=0 )
      WakeUp();
    return push_size;
  }

  bool WorkerSingleQueue::TryPushTask( const TaskPtr& task_ptr ) {
    if(  status() == WorkerQueue::STATUS_LOCKING || status_atomic () ==WorkerQueue::STATUS_RELEASE ) {
        return false;
    }
    task_ptr->set_task_status( Task::TASK_STATUS_WAITING );
    if( !queue_sync_lock_.TryLock() ) {
      return false;
    }
    KillableTask* killable_task_ptr = task_ptr->DownCast<KillableTask>();
    // insert the ptr into the map
    std::pair< QueueType::iterator , bool > ret =queue_.insert(
      QueueType::value_type(killable_task_ptr->name(),task_ptr)
      );
    queue_sync_lock_.Unlock();

    if( !ret.second ) {
      return false;
    } else {
      WakeUp();
      return true;
    }
  }

  bool WorkerSingleQueue::WaitWhenEmpty() {
    // Wait until the empty happened
    while( !empty() ) {
      ThreadManager::SwithToAnotherThread();
    }
    return true;
  }

  bool WorkerSingleQueue::WaitWhenRelease() {
    while( status() != WorkerQueue::STATUS_RELEASE ) {
      ThreadManager::SwithToAnotherThread();
    }
    return true;
  }

  const TripleResult WorkerSingleQueue::WaitWhenEmpty( int ms ) {
    Timer time;
    Timer::tick_t tick;
    if( !time.BeginTimerCounting() ){
      return TripleResult::kException;
    }
    while( !empty() ) {
      tick += time.TimeDiff();
      if( tick > (size_t)(ms) ) 
        ThreadManager::SwithToAnotherThread();
    }
    return TripleResult::kSucceded;
  }

  const TripleResult WorkerSingleQueue::WaitWhenRelease( int ms ) {
    Timer time;
    Timer::tick_t tick;
    if( !time.BeginTimerCounting() ){
      return TripleResult::kException;
    }
    while( status() != WorkerQueue::STATUS_RELEASE ) {
      tick+=time.TimeDiff();
      if( tick > (size_t)(ms) ) 
        ThreadManager::SwithToAnotherThread();
    }
    return TripleResult::kSucceded;
  }

  // Loop for fetching the task and do the execution
  int WorkerSingleQueue::Loop() {
    // Update the status to working
    set_status( WorkerQueue::STATUS_WORKING );
    // task container
    TaskPtrVector task_container;
    TaskPtrVector::iterator ib,ie;
    while( true ) {
      // 1. Testing if there exist a container in the back thread
      if( empty() ) {
        // now we have to sleep now
        set_status( WorkerQueue::STATUS_SLEEP );
        sleep_lock_.Wait();
        set_status( WorkerQueue::STATUS_WORKING );
        // Now we have been waked up.
        // there exit 3 situation that we are waking up 
        // 1. by the push_task function , that means we must fetch task and execute them
        // 2. by the quit_soon 
        // 3. by the quit_empty
        // Dealing process 
        // 1. Wake up by the PostQuitSoon function
        if( quit_option_ == QUIT_SOON ) {
            // just quit now
            set_status_atomic( WorkerQueue::STATUS_RELEASE );
            return WorkerQueue::LOOP_OK_RETURN;
        }
        // 2. Wake up by the PostQuitWhenEmpty function
        if( quit_option_ == QUIT_WHEN_EMPTY ) {
          // When we reaching here 
          // this is possible that we have some work to deal with
          if(!empty()) {
            Dequeue(&task_container,0);
            for( ib = task_container.begin() , ie = task_container.end() ; ib != ie ; ++ib ) {
              (*ib)->set_task_status( Task::TASK_STATUS_RUNNING );
              (*ib)->Execute();
              (*ib)->set_task_status( Task::TASK_STATUS_RELEASE );
            }
          }
          set_status_atomic(WorkerQueue::STATUS_RELEASE);
          return WorkerQueue::LOOP_OK_RETURN;
        }
        continue; // check the empty again 
      }
      // 2. Fetch the task into the container and executing them
      Dequeue(&task_container,precache_size_);
      // executing the task_container here
      for( ib = task_container.begin() , ie = task_container.end() ; ib != ie ; ++ib ) {
        (*ib)->set_task_status( Task::TASK_STATUS_RUNNING );
        (*ib)->Execute();
        (*ib)->set_task_status( Task::TASK_STATUS_RELEASE );
        if( quit_option_ == QUIT_SOON ) {
          // quit now 
          set_status_atomic(WorkerQueue::STATUS_RELEASE);
          return WorkerQueue::LOOP_OK_RETURN;
        }
      }
      // 3. Testing the quit when empty 
      if( quit_option_ == QUIT_WHEN_EMPTY ) {
        Dequeue(&task_container,0);
        for( ib = task_container.begin() , ie = task_container.end() ; ib != ie ; ++ib ) {
          (*ib)->set_task_status( Task::TASK_STATUS_RUNNING );
          (*ib)->Execute();
          (*ib)->set_task_status( Task::TASK_STATUS_RELEASE );
        }
        set_status_atomic(WorkerQueue::STATUS_RELEASE);
        return WorkerQueue::LOOP_OK_RETURN;
      }
    }// while
    UNREACH_BRANCH(_xT("Can not reach here!"));
    return WorkerQueue::LOOP_FAILED_RETURN;
  }

  bool WorkerSingleQueue::KillTask( const String& task_name ) {
    if( status() == WorkerQueue::STATUS_LOCKING || status_atomic() == WorkerQueue::STATUS_RELEASE) 
      return false;
    if( !empty() ) {
      AutoLock<FastLock> l(queue_sync_lock_);
      QueueType::iterator ret = queue_.find(task_name);
      if( ret == queue_.end() ) {
        return false;
      }
      queue_.erase(ret);
      return true;
    }
    return false;
  }

}// namespace xge_internal
}// namespace xge 